package com.example.kafkatesttool;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonSyntaxException;
import javafx.application.Platform;
import javafx.collections.FXCollections;
import javafx.collections.ObservableList;
import javafx.fxml.FXML;
import javafx.fxml.Initializable;
import javafx.scene.control.*;
import javafx.scene.control.cell.PropertyValueFactory;
import javafx.scene.input.ContextMenuEvent;
import javafx.scene.input.MouseButton;
import javafx.scene.input.MouseEvent;
import javafx.stage.FileChooser;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static org.apache.kafka.common.config.SaslConfigs.SASL_KERBEROS_SERVICE_NAME;


public class KafkaController implements Initializable {

    public TreeView<String> treeView;

    @FXML
    private TextField txtServers;

    @FXML
    private TextField txtGroupId;
    @FXML
    private TextField txtTruststore;


    @FXML
    private TextField txtTruststorePassword;

    @FXML
    private TextField txtPrincipal;
    @FXML
    private TextField txtKeyTab;
    @FXML
    private TextField txtKrb5;
    @FXML
    Button btnReceiveMsg;

    @FXML
    Button btnStopMsg;

    @FXML
    private Label labelStatus;

    @FXML
    Label configStatusLabel;

    @FXML
    private Label labelReceiveStatus;

    public ObservableList<KafkaMessage> messageRecords;


    @FXML
    private TextArea sendMesTextArea;

    @FXML
    private TableView<KafkaMessage> tableView;

    @FXML
    private TableColumn<KafkaMessage, String> topicColumn;

    @FXML
    private TableColumn<KafkaMessage, Long> offsetColumn;
    @FXML
    private TableColumn<KafkaMessage, Long> timeStampColumn;

    @FXML
    private TableColumn<KafkaMessage, Integer> partitionColumn;

    @FXML
    private TableColumn<KafkaMessage, String> messageColumn;


    private int maxTopic = 10;

    private int maxConsumers;

    ExecutorService executorService = Executors.newFixedThreadPool(maxTopic);

    private boolean startListen = false;

    ContextMenu contextMenu;

    String processingColor = "#2aeb54";

    String selectTopic;

    private HashMap<String, Boolean> topicListenMap = new HashMap<>();

    TreeItem<String> topicTreeNode = new TreeItem<>();

    HashMap<String, List<TopicPartitionInfo>> topicInfo = new HashMap<>();

    boolean needInit = true;

    private static final ThreadLocal<KafkaProducer<String, String>> threadLocal = new InheritableThreadLocal<>();

    public KafkaProducer<String, String> createProducer(KafkaConfig kafkaConfig) {
        Properties properties = getConnectProperties(kafkaConfig);
        KafkaProducer<String, String> producer;
        if (needInit) {
            producer = new KafkaProducer<>(properties);
        } else {
            producer = threadLocal.get();
            if (Objects.isNull(producer)) {
                producer = new KafkaProducer<>(properties);
            }

        }
        needInit = false;
        threadLocal.set(producer);
        return producer;
    }

    public void cLoseMenu(MouseEvent event) {
        if (contextMenu != null && contextMenu.isShowing() && event.getButton() != MouseButton.SECONDARY) {
            contextMenu.hide();
        }
    }


    private void waring(String warningMsg) {
        Alert alert = new Alert(Alert.AlertType.WARNING);
        alert.setTitle("Warning");
        alert.setHeaderText(warningMsg);
        alert.showAndWait();
    }

    private boolean isBlank(String str) {
        if (str == null || str.length() == 0) {
            return true;
        }

        return false;
    }


    public void showContextMenu(ContextMenuEvent event) {
        TreeItem<String> selectedItem = treeView.getSelectionModel().getSelectedItem();
        if (selectedItem == null) {
            return;
        }
        if (selectedItem.getParent() == null) {
            contextMenu = createContextMenu(selectedItem);
            contextMenu.show(treeView, event.getScreenX(), event.getScreenY());
        } else {
            contextMenu = createDeleteContextMenu(selectedItem);
            contextMenu.show(treeView, event.getScreenX(), event.getScreenY());
        }
    }


    private ContextMenu createContextMenu(TreeItem<String> selectedItem) {
        ContextMenu contextMenu
                = new ContextMenu();
        MenuItem topicItem = new MenuItem("Add Topic");
        topicItem.setOnAction(e -> showAddItemDialog(selectedItem));
        contextMenu.getItems().add(topicItem);

        return contextMenu;

    }

    private void showAddItemDialog(TreeItem<String> selectedItem) {


        TextInputDialog dialog = new TextInputDialog();
        dialog.setTitle("Add Topic");
        dialog.setHeaderText("Enter the Topic name: ");
        dialog.setContentText("Name: ");
        Optional<String> result = dialog.showAndWait();
        result.ifPresent(name -> {
            String nodeName = name.trim();
            if (!nodeName.isEmpty()) {
                if (containsNodeName(selectedItem, name)) {
                    waring("Topic name already exists");
                } else if (topicListenMap.size() >= 1) {
                    waring("Only support one topic at present");

                } else {

                    TreeItem<String> newItem = new TreeItem<>(name);
                    topicListenMap.put(name, false);
                    selectTopic = name;
                    selectedItem.getChildren().add(newItem);
                    treeView.getSelectionModel().select(newItem);
                    topicTreeNode = treeView.getSelectionModel().getSelectedItem();

                }
            } else {
                Alert alert = new Alert(Alert.AlertType.WARNING);
                alert.setTitle("Invalid Input");
                alert.setHeaderText("Topic name cannot be empty");
                alert.showAndWait();
            }
        });
    }


    private ContextMenu createDeleteContextMenu(TreeItem<String> selectedItem) {
        ContextMenu contextMenu = new ContextMenu();
        MenuItem deleteItem = new MenuItem("Delete Topic");
        deleteItem.setOnAction(e -> deleteSelectedItem(selectedItem));
        contextMenu.getItems().add(deleteItem);
        return contextMenu;

    }


    private void deleteSelectedItem(TreeItem<String> selectedItem) {
        if (selectedItem != null) {
            Alert alert = new Alert(Alert.AlertType.CONFIRMATION);
            alert.setTitle("Delete Topic");
            alert.setHeaderText(null);
            alert.setContentText("Are you sure you want to delete the selected topic?");
            Button deleteButton = (Button) alert.getDialogPane().lookupButton(ButtonType.OK);
            deleteButton.setText("Delete");
            Optional<ButtonType> result = alert.showAndWait();
            if (result.isPresent() && result.get() == ButtonType.OK) {
                topicListenMap.remove(selectedItem.getValue());
                selectedItem.getParent().getChildren().remove(selectedItem);
            }
        }
    }

    public void closeMenu(MouseEvent event) {
        if (contextMenu != null && contextMenu.isShowing() && event.getButton() != MouseButton.SECONDARY) {
            contextMenu.hide();
        }
    }

    public void receiveMsg() {
        if (isBlank(selectTopic)) {
            waring("Please add topic you want to Listen");
            return;
        }

        executorService = Executors.newFixedThreadPool(maxTopic);
        startListen = true;

        for (int i = 0; i < topicInfo.get(selectTopic).size();
             i++) {
            executorService.submit(() -> listenTopic(selectTopic));
        }
    }


    public void loadConfig() {
        FileChooser fileChooser = new FileChooser();
        FileChooser.ExtensionFilter extFilter = new FileChooser.ExtensionFilter("JSON files (*.json)", "* . json");
        fileChooser.getExtensionFilters().add(extFilter);
        fileChooser.setSelectedExtensionFilter(extFilter);
        fileChooser.setTitle("Load Config");
        File file = fileChooser.showOpenDialog(null);

        if (file != null) {

            try {
                String jsonContent = new String(Files.readAllBytes(Paths.get(file.getAbsolutePath())));
                Gson gson = new Gson();
                KafkaConfig kafkaConfig = gson.fromJson(jsonContent, KafkaConfig.class);
                txtServers.setText(kafkaConfig.getServers());
                txtPrincipal.setText(kafkaConfig.getPrincial());
                txtGroupId.setText(kafkaConfig.getGroupId());
                txtKeyTab.setText(kafkaConfig.getKeyTabPath());
                txtKrb5.setText(kafkaConfig.getKrb5Path());
                txtTruststore.setText(kafkaConfig.getTrustStorePath());
                txtTruststorePassword.setText(kafkaConfig.getTruststorePassword());
                updateStatus("Load config success", "green", configStatusLabel);
            } catch (JsonSyntaxException | IOException e) {
                System.out.println("read config from file fail-->"
                        + e.getMessage());
                updateStatus("read config from file fail-->" + e.getMessage(),
                        "red", configStatusLabel);
            }
        }
    }

        public void saveConfig() {
            KafkaConfig kafkaConfig = new KafkaConfig();
            kafkaConfig.setServers(txtServers.getText());
            kafkaConfig.setPrincial(txtPrincipal.getText());
            kafkaConfig.setGroupId(txtGroupId.getText());
            kafkaConfig.setKeyTabPath(txtKeyTab.getText());
            kafkaConfig.setKrb5Path(txtKrb5.getText());
            kafkaConfig.setTrustStorePath(txtTruststore.getText());
            kafkaConfig.setTruststorePassword(txtTruststorePassword.getText());
            var gson = new GsonBuilder().setPrettyPrinting().create();
            String content = gson.toJson(kafkaConfig);
            FileChooser fileChooser = new FileChooser();
            FileChooser.ExtensionFilter extFilter = new FileChooser.ExtensionFilter("JSON files (*.json)");
            fileChooser.getExtensionFilters().add(extFilter);

            fileChooser.setSelectedExtensionFilter(extFilter);
            fileChooser.setTitle("Save Config to file");
            File file = fileChooser.showSaveDialog(null);
            if (file != null) {
                try (FileWriter writer = new FileWriter(file)) {
                    writer.write(content);
                } catch (IOException e) {

                    System.out.println("save config->" + e.getMessage());
                    updateStatus("save config->"
                            + e.getMessage(), "red", configStatusLabel);
                }
            }


        }

        @Override
        public void initialize (URL ur, ResourceBundle resourceBundle){
            timeStampColumn.setCellValueFactory(new PropertyValueFactory<>("timeStamp"));
            topicColumn.setCellValueFactory(new PropertyValueFactory<>("topic"));
            partitionColumn.setCellValueFactory(new PropertyValueFactory<>("partition"));
            offsetColumn.setCellValueFactory(new PropertyValueFactory<>("offset"));
            messageColumn.setCellValueFactory(new PropertyValueFactory<>("message"));
            messageRecords = FXCollections.observableArrayList();
            tableView.setItems(messageRecords);
            btnStopMsg.setDisable(true);
        }

        public Properties getConnectProperties (KafkaConfig kafkaConfig){

            // Set the Kerberos configuration file path
            System.setProperty("iava-secunity.Krb5.conf", kafkaConfig.getKrb5Path());
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getServers());
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.

                    getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.

                    getName());
            props.put("security.protocol", "SASL_SSL");
            props.put(SaslConfigs.SASL_MECHANISM, SaslConfigs.GSSAPI_MECHANISM);
            props.put(SASL_KERBEROS_SERVICE_NAME, "Kafka");
            props.put(SaslConfigs.SASL_JAAS_CONFIG, "com.sun.security.auth.module.Krb5LoginModule required\n" +
                    "useKeyTab=true\n" +
                    "KeyTab=)" + kafkaConfig.getKeyTabPath() + "\"\n" +
                    "storeKey=true\n" +
                    "useTicketCache=false\n" +
                    "principal=\"" + kafkaConfig.getPrincial() + "\";");
            props.put("ssl. truststore.location", kafkaConfig.getTrustStorePath());
            props.put("ssl.truststore.password", kafkaConfig.getTruststorePassword());
            props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConfig.getGroupId());
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.

                    getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.

                    getName());
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "Latest");
            props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getServers());
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
            return props;
        }

        private void updateStatus (String msg, String style, Label label){
            Thread updateThread = new Thread(() -> {

                Platform.runLater(() ->
                {
                    label.setText(msg);
                    label.setStyle("-fx-background-color:" + style);


                });
            });
            updateThread.start();
        }


        public void stopListenNsg () {
            if (!startListen) {
                return;
            }
            btnReceiveMsg.setDisable(false);
            btnStopMsg.setDisable(true);
            startListen = false;
            needInit = false;
            threadLocal.remove();
            createProducer(getConfig());
            executorService.shutdown();
            updateStatus("stop receiving message", "#f0e335;", labelReceiveStatus);
        }


        private void listenTopic (String topic){
//        Create the Kafka consumer
            try {
                KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getConnectProperties(getConfig()));
                consumer.subscribe(Collections.singletonList(topic));
                btnReceiveMsg.setDisable(true);
                btnStopMsg.setDisable(false);
                while (startListen) {
                    updateStatus("receiving message", processingColor, labelReceiveStatus);
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> item : records) {
                        int partition = item.partition();
                        Long offset = item.offset();
                        String message = item.value();
                        Date dateTime = new Date(item.timestamp());
                        SimpleDateFormat formatter = new SimpleDateFormat("HH: mm:ss");
                        String timeString = formatter.format(dateTime);
                        messageRecords.add(new KafkaMessage(topic, offset, partition, message, timeString));
                        if (startListen) {
                            updateStatus("message received", "green", labelReceiveStatus);
                        }

                        if (messageRecords.size() > 50) {
                            messageRecords.remove(0);
                        }
                    }

                    System.out.println("123");
                }

                System.out.println("1253546");
            } catch (Exception e) {
                updateStatus("listenTopic fail-->" + e.getMessage(), "red", labelReceiveStatus);
                System.out.println("listenTopic fail-->" + e.getMessage());
            }
        }


        private boolean containsNodeName (TreeItem < String > item, String nodeName){
            if (item.isLeaf() && item.getValue().equals(nodeName)) {
                return true;
            }
            for (TreeItem<String> child : item.getChildren()) {
                if (containsNodeName(child, nodeName))
                    return true;
            }

            return false;
        }


        private KafkaConfig getConfig () {
            KafkaConfig kafkaConfig = new KafkaConfig();
            kafkaConfig.setServers(txtServers.getText());
            kafkaConfig.setPrincial(txtPrincipal.getText());
            kafkaConfig.setKrb5Path(txtKrb5.getText());
            kafkaConfig.setTrustStorePath(txtTruststore.getText());
            kafkaConfig.setTruststorePassword(txtTruststorePassword.getText());

            return kafkaConfig;
        }


        public void getPartitionInfo () {

            if (isBlank(selectTopic)) {
                waring("please add topic");
                return;
            }
            String topicName = selectTopic;
            Properties props = getConnectProperties(getConfig());
            try (AdminClient adminClient = AdminClient.create(props)) {

                DescribeTopicsResult describeResult = adminClient.describeTopics(Collections.singleton(topicName));
                KafkaFuture<TopicDescription> topicDescriptionFuture = describeResult.values().get(topicName);
                TopicDescription topicDescription = topicDescriptionFuture.get();
                topicInfo.put(selectTopic, topicDescription.partitions());
                for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                    TreeItem<String> partintionItem = new TreeItem<>("Partition" + partitionInfo.partition());
                    topicTreeNode.getChildren().add(partintionItem);
                    topicTreeNode.setExpanded(true);
                    updateStatus("Get topic partition success",
                            "green", configStatusLabel);
                }
            } catch (InterruptedException | ExecutionException e) {
                System.out.println("connect kafka fail");
                updateStatus("connect kafka fail" + e.getMessage(), "red", configStatusLabel);

            }


        }
    }